package com.aotain.nyx.icmpflood;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;

import com.aotain.nyx.flink.BasicTuple;


public class ICMPFloodReduce implements ReduceFunction<Tuple2<String, BasicTuple>>{

	/**
	 * 
	 */
	private static final long serialVersionUID = 7011815020769155950L;

	@Override
	public Tuple2<String, BasicTuple> reduce(Tuple2<String, BasicTuple> value1,
			Tuple2<String, BasicTuple> value2) throws Exception {
		// TODO Auto-generated method stub
		
		BasicTuple AandRet = value1.f1;
		BasicTuple B = value2.f1;
		
		AandRet.setUpStreamOctets(AandRet.getUpStreamOctets() + B.getUpStreamOctets());
		AandRet.setUpStreamPacket(AandRet.getUpStreamPacket() + B.getUpStreamPacket());
		
		Tuple2<String, BasicTuple> ret = new Tuple2<String, BasicTuple>();
		ret.setFields(value1.f0, AandRet);
		
		return ret;
	}

}
